package io.gitee.hfl.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.Feature;
import io.gitee.hfl.rocketmq.exception.RocketParseMessageException;
import io.gitee.hfl.rocketmq.msg.RocketMessage;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.StandardCharsets;

/**
 * consumer 需要实现的接口
 *
 * @param <T> 消息体泛型
 * @author hefulin
 * @date 2022/09/07 11:28
 */
public interface RocketListener<T extends RocketMessage> extends MessageListener {

    /**
     * 消费处理逻辑
     *
     * @param message     消息全部内容
     * @param messageBody 消息体
     * @return 处理结果
     */
    ConsumeResult consume(MessageView message, T messageBody);

    /**
     * 消息处理逻辑包装 解析消息体 打印日志
     *
     * @param message 消息全部内容
     * @return 处理结果
     */
    @Override
    default ConsumeResult consume(MessageView message) {
        T messageBody = JSON.parseObject(this.getString(message.getBody()), this.getJsonType().getType(), Feature.SupportAutoType);
        LoggerFactory.getLogger(getClass()).info("consumer received msg:{}", messageBody);
        return consume(message, messageBody);
    }

    /**
     * 获取 json 类型
     *
     * @return 类型
     */
    default TypeReference<T> getJsonType() {
        try {
            Type[] interfaces = getClass().getGenericInterfaces();
            for (Type anInterface : interfaces) {
                if (anInterface.getTypeName().contains(RocketListener.class.getTypeName())) {
                    Type[] actualTypeArguments = ((ParameterizedType) anInterface).getActualTypeArguments();

                    if (actualTypeArguments != null && actualTypeArguments.length > 0) {
                        return new TypeReference<T>() {
                            @Override
                            public Type getType() {
                                return actualTypeArguments[0];
                            }
                        };
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * bytebuffer转string
     *
     * @param buffer 消息体 buffer
     * @return 消息体String
     */
    default String getString(ByteBuffer buffer) {
        CharsetDecoder decoder;
        CharBuffer charBuffer;
        try {
            decoder = StandardCharsets.UTF_8.newDecoder();
            charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
            return charBuffer.toString();
        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RocketParseMessageException("decode byteBuffer error");
        }
    }

}
